#pragma once
#include <carbon/defs.h>
#include <queue>
#include <memory>
#include <mutex>
#include <cassert>
#include <condition_variable>

CARBON_NS_BEGIN


/// @brief  Blocking queue template class
/// @tparam T   element type
template <typename T, typename C = std::queue<T>>
class BlockingQueue { 
    CARBON_NONCOPYABLE(BlockingQueue)
public:
    using Container = C;
    using Element   = T;
    using Lock      = std::mutex;
    using Cond      = std::condition_variable;
    using LockGuard = std::unique_lock<std::mutex>;

public:
    BlockingQueue() = default;
    BlockingQueue(size_t capacity) : m_capacity(capacity) {}
    ~BlockingQueue() = default;

    size_t capacity() const { return m_capacity; }

    /// @brief 向队列写入元素。非阻塞，队列满则立即返回false。
    /// @param element 
    /// @return 
    bool offer(Element&& element) {
        LockGuard lock(m_lock);
        size_t n = m_queue.size();
        if ( n >= m_capacity ) return false;
        m_queue.push(std::move(element));
        if ( n == 0 ) this->m_cond_get.notify_all();
        return true;
    }

    bool offer(const Element& element) {
        LockGuard lock(m_lock);
        size_t n = m_queue.size();
        if ( n >= m_capacity ) return false;
        m_queue.push(element);
        if ( n == 0 ) this->m_cond_get.notify_all();
        return true;
    }

    /// @brief 写入元素, 队列满则阻塞，直至有数据取出，或者超时。超时返回false。
    /// @param element 
    /// @param timeout 
    /// @return 
    bool offer(Element&& element, std::chrono::milliseconds timeout) {
        LockGuard lock(m_lock);
        auto exp = std::chrono::steady_clock::now() + timeout;
        size_t n = m_queue.size();
        while ( n >= this->m_capacity ) {
            auto status = m_cond_put.wait_until(lock, exp);
            if ( status == std::cv_status::timeout) return false;
            else n = m_queue.size();
        }
        
        // 收到可写信号。
        m_queue.push(std::move(element));
        if ( n == 0 ) this->m_cond_get.notify_all();
        return true;
    }

    bool offer(const Element& element, std::chrono::milliseconds timeout) {
        LockGuard lock(m_lock);
        
        auto exp = std::chrono::steady_clock::now() + timeout;
        size_t n = m_queue.size();
        while ( n >= this->m_capacity ) {
            auto status = m_cond_put.wait_until(lock, exp);
            if ( status == std::cv_status::timeout) return false;
            else n = m_queue.size();
        }
        
        // 收到可写信号。
        m_queue.push(element);
        if ( n == 0 ) this->m_cond_get.notify_all();
        return true;
    }

    /// @brief 获取队列头部的元素，不会从队列删除。
    /// @return 
    Optional<Element> peek() const {
        LockGuard lock(m_lock);
        if ( m_queue.size() == 0 ) return Optional<Element>::empty();
        else return Optional<Element>::of(m_queue.front());
    }

    /// @brief 获取队列头部的元素，并从队列删除。不会阻塞。队列空时返回Emtpy.
    /// @return 
    Optional<Element> poll() {
        LockGuard lock(m_lock);
        if ( m_queue.size() == 0 ) return Optional<Element>::empty();

        Element elem(std::move(m_queue.front()));
        m_queue.pop();
        return Optional<Element>::of(std::move(elem));
    }

    Optional<Element> poll(std::chrono::milliseconds timeout) {
        LockGuard lock(m_lock);
        auto exp = std::chrono::steady_clock::now() + timeout;
        size_t n = m_queue.size();
        if ( n == 0 ) {
            auto status = m_cond_get.wait_until(lock, exp);
            if ( status == std::cv_status::timeout) return Optional<Element>::empty();
            else n = m_queue.size();
        }

        Element elem(std::move(m_queue.front()));
        m_queue.pop();
        return Optional<Element>::of(std::move(elem));
    }

    /// @brief 写入一个元素。如果队列满，则阻塞直到写入成功（返回true），或者异常退出(返回false)。
    /// @param element 
    /// @return 
    bool put(Element && element) {
        LockGuard lock(m_lock);
        
        size_t n = m_queue.size();
        while ( n >=  this->m_capacity ) {
            m_cond_put.wait(lock);    // 持续等待
            n = m_queue.size();
        }
        
        // 收到可写信号。
        m_queue.push(std::move(element));
        if ( n == 0 ) this->m_cond_get.notify_one();
        return true;
    }

    bool put(const Element & element) {
        LockGuard lock(m_lock);
        
        size_t n = m_queue.size();
        while ( n >= this->m_capacity ) {
            m_cond_put.wait(lock);    // 持续等待
            n = m_queue.size();
        }
        
        // 收到可写信号。
        m_queue.push(element);
        if ( n == 0 ) this->m_cond_get.notify_one();
        return true;
    }

    /// @brief 删除队列头部元素. 空队列则立即返回。
    void remove() {
        LockGuard lock(m_lock);
        if ( m_queue.size() == 0 ) return ;
        else m_queue.pop();
    }

    /// @brief 获取头部元素并从队列移除。队列空则阻塞直到有数据可返回。
    /// @return 
    Optional<Element> take() {
        LockGuard lock(m_lock);

        size_t n = m_queue.size();
        if ( n == 0 ) {
            auto status = m_cond_get.wait(lock);
            n = m_queue.size();
        }

        Element elem(std::move(m_queue.front()));
        m_queue.pop();
        return Optional<Element>::of(std::move(elem));
    }

private:
    size_t      m_capacity  {UINT32_MAX};
    Lock        m_lock;         // 访问锁，默认单锁，并发性能一般
    Cond        m_cond_put;     // 可写条件
    Cond        m_cond_get;     // 可读条件
    Container   m_queue;
}; // BlockingQueue

CARBON_NS_END
